package com.doit.mr.day04.join;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @DATE 2021/12/9/22:31
 * @Author MDK
 * @Version 2021.2.2
 **/
public class Join {
    //输出的是 K:uid  V:行内容
    static class JoinMapper extends Mapper<LongWritable, Text, Text, Text>{
        String fileName = null;
        //当前task仅执行一次
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();
        }
        Text k = new Text();
        Text v = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String uid = null;
            //处理数据  根据文件名判断处理的是用户数据还是订单数据
            if(fileName.startsWith("user")){
                String[] split = line.split(",");
                uid = split[0];
            }else {
                String[] split = line.split("\\s+");
                uid = split[1];
            }
            k.set(uid);
            v.set(line);
            context.write(k,v);
        }
    }

    static class JoinReducer extends Reducer<Text, Text, Text, NullWritable>{
        Text k = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            String user = null;
            List<String> orders = new ArrayList<>();
            //遍历所有内容
            for (Text value : values) {
                String content = value.toString();
                //判断内容是否包含用户信息
                if (content.contains(",")){
                    user = content;
                }else{
                    //将订单信息存储在集合中
                    orders.add(content);
                }
            }
            //遍历订单 拼接用户
            for (String order : orders) {
                String join = order.split("\\s+")[0] + "," + user;
                k.set(join);
                context.write(k,NullWritable.get());
            }
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "join");
        job.setMapperClass(JoinMapper.class);
        job.setReducerClass(JoinReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, new Path("D:\\mrdata\\join\\input"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\mrdata\\join\\output"));
        job.waitForCompletion(true);
    }
}
